package reduce;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.math.BigDecimal;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import plagiarismChecker.MapReduceSkeleton;

public class ReduceJob2 extends Reducer<Text, Text, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    	
    	int sum = 0;
    	
    	//leian esmalt euclidean normi
    	List<String> valuesList = new ArrayList<String>();
    	for (Text value : values) {
    		String phrase = value.toString();
    		valuesList.add(phrase);
        	String [] phraseandvalue = phrase.split(MapReduceSkeleton.separator);
        	int count = Integer.parseInt(phraseandvalue[1]);
        	sum += count*count;
    	}
    	
    	BigDecimal euclidean = new BigDecimal(Math.sqrt(sum));
    	
    	//teen iga fraasi kohta contexti
    	for (String value : valuesList) {
    		String [] phraseandvalue = value.split(MapReduceSkeleton.separator);
    		context.write(new Text(phraseandvalue[0]), new Text(key + MapReduceSkeleton.separator + phraseandvalue[1] + MapReduceSkeleton.separator + euclidean));
    	}
    }
}